package com.boilermaker.mq.mqserver.manager;

import com.boilermaker.mq.common.BinaryTool;
import com.boilermaker.mq.common.MqException;
import com.boilermaker.mq.mqserver.core.MSGQueue;
import com.boilermaker.mq.mqserver.core.Message;
import lombok.extern.slf4j.Slf4j;

import java.io.*;
import java.util.LinkedList;

/*
 * 通过这个类管理队列的消息文件
 */
@Slf4j
public class MessageFileManager {

    // 该静态内部类用于描述队列中消息的统计信息
    static public class Stat {
        public int totalCount; // 总消息数
        public int validCount; // 有效消息数
    }

    // 通过队列名，获取相应的消息文件目录
    private String getQueueDir(String queueName) {
        return "./data/" + queueName;
    }

    // 通过队列名，获取相应的消息文件下的消息数据文件
    private String getQueueDataPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data.bin";
    }

    // 通过队列名，获取相应的消息文件下的消息统计信息文件
    private String getQueueStatPath(String queueName) {
        return getQueueDir(queueName) + "/queue_stat.bin";
    }

    // 获取某队列消息的统计信息
    private Stat readStat(String queueName) {
        Stat stat = new Stat();
        try (FileInputStream fis = new FileInputStream(getQueueStatPath(queueName));
             DataInputStream dis = new DataInputStream(fis)) {
            stat.totalCount = dis.readInt();
            stat.validCount = dis.readInt();
            return stat;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    // 更新某队列消息的统计信息
    private void writeStat(String queueName, Stat stat) {
        // OutputStream 默认使用新数据覆盖旧数据
        try (FileOutputStream fos = new FileOutputStream(getQueueStatPath(queueName));
             DataOutputStream dos = new DataOutputStream(fos)) {
            dos.writeInt(stat.totalCount);
            dos.writeInt(stat.validCount);
            dos.flush();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    // 封装创建目录和文件的过程
    public void createQueueFile(String queueName) throws IOException {
        // 创建消息文件目录
        File queueDir = new File(getQueueDir(queueName));
        if (!queueDir.exists()) {
            boolean ok = queueDir.mkdirs();
            if (!ok) {
                throw new IOException("创建目录失败, queueDir=" + queueDir.getAbsolutePath());
            }
        }

        // 创建消息数据文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            boolean ok = queueDataFile.createNewFile();
            if (!ok) {
                throw new IOException("创建消息数据文件失败, queueDataFile=" + queueDataFile.getAbsolutePath());
            }
        }

        // 创建消息统计信息文件
        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()) {
            boolean ok = queueStatFile.createNewFile();
            if (!ok) {
                throw new IOException("创建消息统计信息文件失败, queueStatFile=" + queueStatFile.getAbsolutePath());
            }
        }

        // 给消息统计信息文件设置初始值
        Stat stat = new Stat();
        stat.totalCount = 0;
        stat.validCount = 0;
        writeStat(queueName, stat);
    }

    // 删除队列的数据文件和目录
    public void deleteQueueFile(String queueName) throws IOException {
        // 先删文件，再删目录
        File queueDataFile = new File(getQueueDataPath(queueName));
        boolean ok1 = queueDataFile.delete();
        File queueStatFile = new File(getQueueStatPath(queueName));
        boolean ok2 = queueStatFile.delete();
        File queueDir = new File(getQueueDir(queueName));
        boolean ok3 = queueDir.delete();
        if (!ok1 || !ok2 || !ok3) throw new IOException("删除文件或目录失败。queueDir=" + queueDir.getAbsolutePath());
    }

    // 检查队列的数据文件是否存在
    public boolean checkQueueFileExist(String queueName) {
        File queueDataFile = new File(getQueueDataPath(queueName));
        File queueStatFile = new File(getQueueStatPath(queueName));
        return queueStatFile.exists() && queueDataFile.exists();
    }

    // 将一个消息写入对应的队列文件，并更新统计文件
    public void sendMessage(MSGQueue queue, Message message) throws MqException, IOException {
        // 检查该文件是否存在
        if (!checkQueueFileExist(queue.getName())) {
            throw new MqException("队列对应的文件不存在，queueName=" + queue.getName());
        }
        // 序列化
        byte[] messageBinary = BinaryTool.toBytes(message);
        // 保证针对同一文件的更新是串行执行，以队列作为锁对象，多个线程操作同一个队列则会阻塞
        synchronized (queue) {
            // 操作文件
            File queueDataFile = new File(getQueueDataPath(queue.getName()));
            // 计算 offsetBegin 和 offsetEnd，更新到 Message 对象中，后续可以通过这两个属性取出消息
            message.setOffsetBegin(queueDataFile.length() + 4);
            message.setOffsetEnd(queueDataFile.length() + 4 + messageBinary.length);
            // 在文件中追加写
            // 使用 DataOutputStream 包一层，便于写入 int
            try (FileOutputStream fileOutputStream = new FileOutputStream(queueDataFile, true);
                 DataOutputStream dataOutputStream = new DataOutputStream(fileOutputStream)) {
                dataOutputStream.writeInt(messageBinary.length);
                dataOutputStream.write(messageBinary);
            }
            // 更新统计文件信息
            Stat stat = readStat(queue.getName());
            assert stat != null;
            stat.totalCount++;
            stat.validCount++;
            writeStat(queue.getName(), stat);
        }
    }

    // 将一个消息从对应的队列文件删除（逻辑删除，即将文件中该消息占有的位置标记为无效），并更新统计文件
    public void deleteMessage(MSGQueue queue, Message message) throws IOException, ClassNotFoundException {
        // 保证针对同一文件的更新是串行执行，以队列作为锁对象，多个线程操作同一个队列则会阻塞
        synchronized (queue) {
            // 该流对象允许调用者从文件的任意位置开始写入或读取
            try (RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(queue.getName()), "rw")) {
                // 空数组用于接收读取出来的数据
                byte[] buffer = new byte[(int) (message.getOffsetEnd() - message.getOffsetBegin())];
                // seek 方法用于移动文件光标到指定位置
                randomAccessFile.seek(message.getOffsetBegin());
                // 读取数据
                randomAccessFile.read(buffer);
                // 将数据反序列化为 Java 对象
                Message diskMessage = (Message) BinaryTool.fromBytes(buffer);
                // 逻辑删除
                diskMessage.setIsValid((byte) 0x0);
                // Java 对象相当于一个过渡，只在边界处存在，做完逻辑删除的标记后，立即将其序列化，写回文件
                // 写文件之前仍需确保光标在正确的位置（调用 read 后，光标位置已经改变了）
                randomAccessFile.seek(message.getOffsetBegin());
                randomAccessFile.write(BinaryTool.toBytes(diskMessage));
            }
            // 更新统计文件
            Stat stat = readStat(queue.getName());
            assert stat != null;
            if (stat.validCount > 0) stat.validCount--;
            writeStat(queue.getName(), stat);
        }
    }

    // 从文件中读取全部有效消息，加载到内存中
    // 使用链表方便头删
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException, ClassNotFoundException, MqException {
        LinkedList<Message> messageList = new LinkedList<>();
        try (FileInputStream inputStream = new FileInputStream(getQueueDataPath(queueName));
             DataInputStream dataInputStream = new DataInputStream(inputStream)) {
            // 手动记录文件光标位置
            long currentOffset = 0;
            while (true) {
                // 读取文件长度，读取到文件末尾时该方法会抛出异常
                int messageSize = dataInputStream.readInt();
                currentOffset += 4;
                // 读取文件内容
                byte[] buffer = new byte[messageSize];
                int actualSize = dataInputStream.read(buffer);
                if (actualSize != messageSize) {
                    throw new MqException("文件格式有误，queueName=" + queueName);
                }
                currentOffset += messageSize;
                // 反序列化
                Message diskMessage = (Message) BinaryTool.fromBytes(buffer);
                // 无效消息直接忽略
                if (diskMessage.getIsValid() != 0x1) {
                    continue;
                }
                // 更新并记录偏移量
                diskMessage.setOffsetBegin(currentOffset - messageSize);
                diskMessage.setOffsetEnd(currentOffset);
                // 添加至链表
                messageList.add(diskMessage);
            }
        } catch (EOFException e) { // 该异常为程序运行中的正常情况，表示文件已读完
            log.info("恢复 Message 数据完成");
        }
        return messageList;
    }

    // 判定某队列文件是否需要进行垃圾回收
    public Boolean checkGC(String queueName) {
        Stat stat = readStat(queueName);
        assert stat != null;
        return stat.totalCount > 2000 && (double) stat.validCount / (double) stat.totalCount < 0.5;
    }

    // 将旧文件中的有效数据移动至此处
    private String getQueueDataNewPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data_new.bin";
    }

    // 执行垃圾回收操作
    public void gc(MSGQueue queue) throws MqException, IOException, ClassNotFoundException {
        synchronized (queue) {
            String name = queue.getName();
            // GC 可能比较耗时，需要统计执行时间
            long gcBeg = System.currentTimeMillis();
            // 操作文件
            File queueDataNewFile = new File(getQueueDataNewPath(name));
            // 该文件若已经存在，则说明某次调用出现逻辑错误，导致出现残留，此时抛出异常
            if (queueDataNewFile.exists()) {
                throw new MqException("垃圾回收时已存在目标文件");
            }
            // 创建新文件
            boolean ok = queueDataNewFile.createNewFile();
            if (!ok) {
                throw new MqException("创建文件失败，queueDataNewFile=" + queueDataNewFile.getAbsolutePath());
            }
            // 从旧文件中读取全部有效消息，加载到内存中
            LinkedList<Message> messageList = loadAllMessageFromQueue(name);
            // 将消息写回新文件
            try (FileOutputStream outputStream = new FileOutputStream(queueDataNewFile);
                 DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                for (Message message : messageList) {
                    byte[] buffer = BinaryTool.toBytes(message);
                    dataOutputStream.writeInt(buffer.length);
                    dataOutputStream.write(buffer);
                }
            }
            // 删除旧的数据文件，并将新文件重命名
            File queueDataOldFile = new File(getQueueDataPath(name));
            ok = queueDataOldFile.delete();
            if (!ok) {
                throw new MqException("删除旧数据文件失败，queueDataOldFile=" +
                        queueDataOldFile.getAbsolutePath());
            }
            ok = queueDataNewFile.renameTo(queueDataOldFile);
            if (!ok) {
                throw new MqException("文件重命名失败，queueDataOldFile=" +
                        queueDataOldFile.getAbsolutePath() + "，queueDataNewFile=" + queueDataNewFile.getAbsolutePath());
            }
            // 更新统计文件
            Stat stat = readStat(name);
            assert stat != null;
            stat.totalCount = messageList.size();
            stat.validCount = messageList.size();
            // 统计用时
            long gcEnd = System.currentTimeMillis();
            log.info("垃圾回收完成，queueName=" + name + "，time=" + (gcEnd - gcBeg) + "ms");
        }
    }
}
